package KafkaTest

import (
	"fmt"
	"github.com/Shopify/sarama"
	cluster "github.com/bsm/sarama-cluster"
	"github.com/robfig/cron"
	"log"
	"time"
)

var (
	KafkaConsumer *cluster.Consumer
	times         = 0
)

func ConsumerInit() {

	var err error
	config := cluster.NewConfig()
	config.Consumer.Return.Errors = true
	config.Group.Return.Notifications = true
	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
	config.Consumer.Offsets.Initial = -2
	config.Consumer.Offsets.CommitInterval = 1 * time.Second
	config.Group.Return.Notifications = true
	KafkaConsumer, err = cluster.NewConsumer(KafkaConfigs.KafkaBrokers, KafkaConfigs.GroupId, []string{KafkaConfigs.KafkaTopic}, config)
	if err != nil {
		log.Printf("Kafka消费者程序异常！-> %v \n", err)
		panic(err.Error())
	}
	if KafkaConsumer == nil {
		panic(fmt.Sprintf("Kafka消费者初始化异常。 Kafka配置 -> {brokers:%v, topic: %v, group: %v}", KafkaConfigs.KafkaBrokers, KafkaConfigs.KafkaTopic, KafkaConfigs.GroupId))
	}
	log.Printf("Kafka消费者初始化成功。消费者配置 -> %v, topic -> %v, \n", KafkaConsumer, KafkaConfigs.KafkaTopic)
	consmerCron()
}

func consmerCron() {
	log.Println("Starting Cron...")

	c := cron.New()
	spec := "*/5 * * * * ?"
	_ = c.AddFunc(spec, consumerWork)
	c.Start()
	select {} //阻塞主线程停止
}

func consumerWork() {
	times++
	log.Printf("第%v次扫描topic", times)
	for {
		select {
		case msg, ok := <-KafkaConsumer.Messages():
			if ok {
				log.Printf("Kafka消费者: %v \n", msg)
				log.Printf("Kafka消费者获取到的信息: %s \n", msg.Value)
				KafkaConsumer.MarkOffset(msg, "")
			} else {
				log.Printf("Kafka监听服务失败 \n")
			}
		case err, ok := <-KafkaConsumer.Errors():
			if ok {
				log.Printf("Kafka消费者异常: %v \n", err)
			}
		case ntf, ok := <-KafkaConsumer.Notifications():
			if ok {
				log.Printf("Kafka消费者提醒: %v \n", ntf)
			}
		}
	}
}
